package com.lry;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Data
public class MappedFile {

    //操作系统页缓存大小4kb
    public static final int OS_PAGE_SIZE = 4 * 1024;

    //提交指针
    private final AtomicInteger committedPosition = new AtomicInteger(0);

    //写指针
    private final AtomicInteger wrotePosition = new AtomicInteger(0);

    //刷盘指针
    private final AtomicInteger flushedPosition = new AtomicInteger(0);

    //文件大小 字节
    private int fileSize;

    //开启堆外内存：flushedPosition <= committedPosition <= wrotePosition <= fileSize
    //1 <= 3 <= 5 <= 10
    //申请了10MB文件，目前写到5MB处，前3MB文件已从堆外内存刷到pagecache，前1MB被刷到磁盘了
    //写入->堆外内存->pagecache->磁盘

    //不开启堆外内存：flushedPosition <= wrotePosition <= fileSize
    //写入->pagecache->磁盘
    //申请一个10MB的文件，那么fileSize就是10MB, wrotePosition就是当前文件写到哪个位置了，flushedPosition是指文件哪些部分被刷到磁盘了

    private FileChannel fileChannel;

    //是否开启堆外缓存 减轻 page cache的压力
    private ByteBuffer writeBuffer;
    private TransientStorePool transientStorePool;

    private String fileName;

    private long fileFromOffset;

    private File file;

    //page cache mmap
    private MappedByteBuffer mappedByteBuffer;

    private volatile long storeTimestamp;

    private boolean firstCreateInQueue;

    public MappedFile(final String fileName, final int fileSize) throws IOException {
        this(fileName,fileSize,null);
    }

    public MappedFile(final String fileName, final int fileSize,
                      final TransientStorePool transientStorePool) throws IOException {
        this.fileName = fileName;
        this.fileSize = fileSize;
        this.file = new File(fileName);
        this.fileFromOffset = Long.parseLong(this.file.getName());
        boolean ok = false;

        ensureDirOK(this.file.getParent());
        try {
            this.fileChannel = new RandomAccessFile(this.file,"rw").getChannel();
            this.mappedByteBuffer = this.fileChannel.map(FileChannel.MapMode.READ_WRITE,0,fileSize);
            ok = true;
        } catch (FileNotFoundException e) {
            log.error("Failed to create file " + this.fileName, e);
            throw e;
        } catch (IOException e) {
            log.error("Failed to map file " + this.fileName, e);
            throw e;
        } finally {
            if (!ok && this.fileChannel != null) {
                this.fileChannel.close();
            }
        }

        if(null!=transientStorePool){
            this.writeBuffer = transientStorePool.borrowBuffer();
            this.transientStorePool = transientStorePool;
        }
    }

    public static void ensureDirOK(final String dirName) {
        if (dirName != null) {
            File f = new File(dirName);
            if (!f.exists()) {
                boolean result = f.mkdirs();
                log.info(dirName + " mkdir " + (result ? "OK" : "Failed"));
            }
        }
    }

    public long getLastModifiedTimestamp() {
        return this.file.lastModified();
    }

    public boolean append(final byte[] data){
       return append(data,0,data.length);
    }

    public boolean append(final byte[] data,final int offset,final int len){

        if(null==data||data.length==0||offset<0||len>data.length){
            return false;
        }

        int curPos = this.wrotePosition.get();

        if((curPos + len) <= this.fileSize){
            try{
                ByteBuffer buf = this.writeBuffer!=null ?this.writeBuffer.slice():this.mappedByteBuffer.slice();
                buf.position(curPos);
                buf.put(data,offset,len);
            }catch (Throwable e){
                log.error("Error occurred when append message to mappedFile.", e);
            }
            this.wrotePosition.addAndGet(data.length);
            return true;
        }
        return false;
    }

    public int flush(final int flushLeastPages){
        if(this.isAbleToFlush(flushLeastPages)){
            //使用了堆外内存=committedPos，否则=wrotePos
            int value = getReadPosition();

            try{
                if(writeBuffer!=null && fileChannel.position()!=0){
                    //使用了堆外内存，则使用send file 刷盘
                    this.fileChannel.force(false);
                }else{
                    //page cache 刷盘
                    this.mappedByteBuffer.force();
                }
            }catch (Throwable e){
                log.error("Error occurred when force data to disk.", e);
            }
            //set flush position
            this.flushedPosition.set(value);
        }
        return this.getFlushedPosition();
    }

    //从堆外内存刷到page cache
    public int commit(final int commitLeastPages){
        if(this.writeBuffer == null){
            return this.wrotePosition.get();
        }

        if(this.isAbleToCommit(commitLeastPages)){
            commit0();
        }

        if(this.writeBuffer!=null && this.transientStorePool !=null && this.fileSize == this.getCommittedPosition()){
            this.transientStorePool.returnBuffer(writeBuffer);
            this.writeBuffer = null;
        }
        return this.getCommittedPosition();
    }


    private boolean isAbleToCommit(final int commitLeastPages){
        int commit = this.committedPosition.get();
        int write = this.wrotePosition.get();

        if (this.isFull()) {
            return true;
        }

        if (commitLeastPages > 0) {
            return ((write / OS_PAGE_SIZE) - (commit / OS_PAGE_SIZE)) >= commitLeastPages;
        }

        return write > commit;
    }

    private void commit0(){
        int writePos = getWrotePosition();
        int lastCommittedPos = getCommittedPosition();

        if(writePos-lastCommittedPos>0){
            try{
                //使用了堆外内存
                ByteBuffer slice = this.writeBuffer.slice();
                slice.position(lastCommittedPos);
                slice.limit(writePos);
                //使用send file 写到堆外内存，并设置提交指针

                this.fileChannel.position(lastCommittedPos);
                this.fileChannel.write(slice);
                this.committedPosition.set(writePos);
            }catch (Throwable e){
                log.error("Error occurred when commit data to FileChannel.", e);
            }
        }

    }

    public Integer getCommittedPosition() {
        return committedPosition.get();
    }

    public Integer getWrotePosition() {
        return wrotePosition.get();
    }

    public Integer getFlushedPosition(){
        return flushedPosition.get();
    }

    private boolean isAbleToFlush(final int flushLeastPages){
        int flush = this.flushedPosition.get();

        int write = getReadPosition();

        if(isFull()){
            return true;
        }
        if(flushLeastPages>0){
            return ((write/OS_PAGE_SIZE)-(flush/OS_PAGE_SIZE)) >= flushLeastPages;
        }
        return write > flush;
    }

    public boolean isFull() {
        return this.fileSize == this.wrotePosition.get();
    }

    public int getReadPosition() {
        return this.writeBuffer!=null ?committedPosition.get():wrotePosition.get();
    }

    public SelectMappedBufferResult select(final int pos){
        int readPosition = this.getReadPosition();
        return select(pos,readPosition-pos);
    }


    public SelectMappedBufferResult select(final int pos,final int size){
        if(pos<0){
            return null;
        }
        int readPosition = this.getReadPosition();

        if(pos+size > readPosition){
            return null;
        }

        ByteBuffer slice = this.mappedByteBuffer.slice();
        slice.position(pos);

        //slice，positi
        ByteBuffer bufNew = slice.slice();
        bufNew.limit(size);
        return new SelectMappedBufferResult(this.fileFromOffset+pos,bufNew,size,this);
    }


    public long calCurCommitLogOffset(){
        return fileFromOffset +getReadPosition();
    }

    public void release() {
    }

    public void setWrotePosition(int pos) {
        this.wrotePosition.set(pos);
    }


    public void setFlushedPosition(int pos) {
        this.flushedPosition.set(pos);
    }

    public void setCommittedPosition(int pos) {
        this.committedPosition.set(pos);
    }
}
